package com.zhaiyz.grpcdemo

import com.google.common.util.concurrent.FutureCallback
import com.google.common.util.concurrent.Futures.addCallback
import com.zhaiyz.grpcdemo.proto.GreetingRequest
import com.zhaiyz.grpcdemo.proto.GreetingResponse
import com.zhaiyz.grpcdemo.proto.GreetingServiceGrpc
import io.grpc.ManagedChannelBuilder
import io.grpc.stub.StreamObserver
import java.util.concurrent.TimeUnit
import java.util.stream.Stream

/**
 * Created by zhaiyz on 2018/1/17.
 */
fun main(args: Array<String>) {
    val channel = ManagedChannelBuilder.forAddress("localhost", 8080).usePlaintext(true).build()
    val blockStub = GreetingServiceGrpc.newBlockingStub(channel)
    val futureStub = GreetingServiceGrpc.newFutureStub(channel)
    val stub = GreetingServiceGrpc.newStub(channel)

    val request = GreetingRequest.newBuilder().setName("zhaiyz").build()


    block(blockStub, request)

    future(futureStub, request)

    async(stub, request)

    streamRequest(stub)

    streamResponse(stub, request)

    streamOfStream(stub)


    channel.awaitTermination(5, TimeUnit.SECONDS)
}

fun block(blockStub: GreetingServiceGrpc.GreetingServiceBlockingStub, request: GreetingRequest) {
    val start = System.currentTimeMillis()
    val response = blockStub.greeting(request)
    println(response.msg)
    val end = System.currentTimeMillis()
    println("BlockStub=(${end - start})")
}

fun future(futureStub: GreetingServiceGrpc.GreetingServiceFutureStub, request: GreetingRequest) {
    val start = System.currentTimeMillis()
    val future = futureStub.greeting(request)
    addCallback(future, object : FutureCallback<GreetingResponse> {
        override fun onFailure(t: Throwable) {
        }

        override fun onSuccess(result: GreetingResponse?) {
            println(result?.msg)
        }
    })
    val end = System.currentTimeMillis()
    println("FutureStub=(${end - start})")
}

fun async(stub: GreetingServiceGrpc.GreetingServiceStub, request: GreetingRequest) {
    val start = System.currentTimeMillis()
    stub.greeting(request, object : StreamObserver<GreetingResponse> {
        override fun onNext(value: GreetingResponse) {
            println(value.msg)
        }

        override fun onCompleted() {
        }

        override fun onError(t: Throwable) {
        }

    })
    val end = System.currentTimeMillis()
    println("AsyncStub=(${end - start})")
}

fun streamRequest(stub: GreetingServiceGrpc.GreetingServiceStub) {
    val streamRequest = stub.streamGreeting(object : StreamObserver<GreetingResponse> {
        override fun onError(t: Throwable) {
        }

        override fun onNext(value: GreetingResponse) {
            println(value.msg)
        }

        override fun onCompleted() {
        }

    })

    Stream.of("zhaiyz", "zhaiy", "zhai")
            .map { GreetingRequest.newBuilder().setName(it).build() }
            .forEach { streamRequest.onNext(it) }
    streamRequest.onCompleted()
}

fun streamResponse(stub: GreetingServiceGrpc.GreetingServiceStub, request: GreetingRequest) {
    stub.greetingStream(request, object : StreamObserver<GreetingResponse> {
        override fun onNext(value: GreetingResponse) {
            println(value.msg)
        }

        override fun onError(t: Throwable) {
        }

        override fun onCompleted() {
        }
    })
}

fun streamOfStream(stub: GreetingServiceGrpc.GreetingServiceStub) {
    val streamRequest = stub.streamGreetingStream(object : StreamObserver<GreetingResponse> {
        override fun onError(t: Throwable) {
        }

        override fun onCompleted() {
        }

        override fun onNext(value: GreetingResponse) {
            println(value.msg)
        }
    })

    Stream.of("zhaiyz", "zhaiy", "zhai")
            .map { GreetingRequest.newBuilder().setName(it).build() }
            .forEach {
                streamRequest.onNext(it)
                Thread.sleep(2000)
            }
    streamRequest.onCompleted()
}